package david.java.flink.join.dataSource;

import david.java.flink.join.entity.Promotion;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 10:22 上午 2021/11/8
 */
public class ProDataSource implements SourceFunction<Promotion> {

    private volatile boolean isRunning = true;
    private static BufferedReader br;
    private static Integer SLEEP = 1000;
    private static Integer MAX_OUT_OF_ORDERNESS = 10 * 1000;


    public ProDataSource() throws FileNotFoundException {
        File file;
        br = new BufferedReader(new FileReader("src/main/resources/ProData"));
    }

    @Override
    public void run(SourceContext<Promotion> ctx) throws Exception {
        String line = null;
        while (null != (line = br.readLine())) {
            String[] split = line.split(",");
            Promotion pro = new Promotion(Integer.valueOf(split[0]), split[1], Double.valueOf(split[2]));
            Long timestamp = Long.valueOf(split[3]);
            pro.setTimestamp(timestamp);
            ctx.collectWithTimestamp(pro, timestamp);
            ctx.collect(pro);
            ctx.emitWatermark(new Watermark(timestamp - MAX_OUT_OF_ORDERNESS));
            Thread.sleep(SLEEP);
        }

    }

    @Override
    public void cancel() {

    }
}
